package com.lagou.hbase.client;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

public class HBaseClient {
    Configuration conf = null;
    Connection conn = null;
    HBaseAdmin admin = null;

    @Before
    public void init() throws IOException {
        conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "linux121,linux122");
        conf.set("hbase.zookeeper.property.clientPort", "2181");
        conn = ConnectionFactory.createConnection(conf);
    }

    @After
    public void destroy() {
        if (admin != null) {
            try {
                admin.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        if (conn != null) {
            try {
                conn.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }


    @Test
    public void createTable() throws IOException {
        admin = (HBaseAdmin) conn.getAdmin();
        //创建表描述器器
        HTableDescriptor relation = new HTableDescriptor(TableName.valueOf("relation"));
        //设置列族描述器器
        relation.addFamily(new HColumnDescriptor("friends"));
        //执⾏创建操作
        admin.createTable(relation);
        System.out.println("用户关系表创建成功！！");
    }


    //插⼊数据
    @Test
    public void putData() throws IOException {
        //获取⼀一个表对象
        Table t = conn.getTable(TableName.valueOf("relation"));
        //插入数据 uid1,uid2,uid3互为好友
        List<Put> putList = new ArrayList<Put>();
        //设定rowKey
        Put put1 = new Put(Bytes.toBytes("uid1"));
        //列族，列名， value
        put1.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("uid2"), Bytes.toBytes("uid2"));
        put1.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("uid3"), Bytes.toBytes("uid3"));
        putList.add(put1);
        Put put2 = new Put(Bytes.toBytes("uid2"));
        put2.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("uid1"), Bytes.toBytes("uid1"));
        put2.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("uid3"), Bytes.toBytes("uid3"));
        putList.add(put2);
        Put put3 = new Put(Bytes.toBytes("uid3"));
        put3.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("uid1"), Bytes.toBytes("uid1"));
        put3.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("uid2"), Bytes.toBytes("uid2"));
        putList.add(put3);
        //执⾏行插⼊
        t.put(putList);
        //关闭table对象
        t.close();
        System.out.println("插⼊成功！！ ");
    }

    /**
     * 全表扫描
     */
    @Test
    public void scanAllData() throws IOException {
        HTable relation = (HTable) conn.getTable(TableName.valueOf("relation"));
        Scan scan = new Scan();
        ResultScanner resultScanner = relation.getScanner(scan);
        for (Result result : resultScanner) {
            Cell[] cells = result.rawCells();//获取该行的所有cell对象
            for (Cell cell : cells) {
                //通过cell获取rowKey, columnFamily, columnValue
                String columnFamily = Bytes.toString(CellUtil.cloneFamily(cell));
                String value = Bytes.toString(CellUtil.cloneValue(cell));
                String rowKey = Bytes.toString(CellUtil.cloneRow(cell));
                System.out.println(rowKey + "----" + columnFamily + "----" + value);
            }
        }
        relation.close();
    }


}
